package cn.xinfei.xdecision.service;

import cn.xinfei.xdecision.common.utils.util.JsonUtils;
import cn.xinfei.xdecision.conf.ApplicationContextUtil;
import cn.xinfei.xdecision.pulsar.consumer.XPulsarConsumerRegister;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.client.api.ConsumerStats;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class PulsarService {

    public ConsumerStats stats(String topic) {
        XPulsarConsumerRegister XPulsarConsumerRegister = ApplicationContextUtil.getBean("XPulsarConsumerRegister", XPulsarConsumerRegister.class);
        ConsumerStats stats = XPulsarConsumerRegister.stats(topic);
        log.info("pulsar consumer stats={}", JsonUtils.toJsonPrettyString(stats));
        return stats;
    }

    public void restart(String topic) {

        XPulsarConsumerRegister XPulsarConsumerRegister = ApplicationContextUtil.getBean("XPulsarConsumerRegister", XPulsarConsumerRegister.class);
        XPulsarConsumerRegister.startPulsarConsumer(topic);

    }

    public void stop(String topic) {

        XPulsarConsumerRegister XPulsarConsumerRegister = ApplicationContextUtil.getBean("XPulsarConsumerRegister", XPulsarConsumerRegister.class);
        XPulsarConsumerRegister.stop(topic);
    }
}
